
import com.zuikaku.util.TimeUtil;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.util.List;
import java.util.Map;

public class LoginMonitorCEPDemo {
    public static void main(String[] args) {
        //1.创建环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.createLocalEnvironment();
        //2.设置模式和并行数
        environment.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        environment.setParallelism(1);
        //3.指定socket source
        DataStreamSource<String> socketTextDS = environment.socketTextStream("192.168.123.8", 8888);
        //4.转化为Tuple3方便之后的处理
        SingleOutputStreamOperator<Tuple3<Integer, String, Integer>> mapDS = socketTextDS.map(new MapFunction<String, Tuple3<Integer, String, Integer>>() {
            @Override
            public Tuple3<Integer, String, Integer> map(String input) throws Exception {
                String[] strArr = input.split(",");
                return Tuple3.of(Integer.parseInt(strArr[0]),strArr[1],Integer.parseInt(strArr[2]));
            }
        });
        //5.指定事件时间列
        SingleOutputStreamOperator<Tuple3<Integer, String, Integer>> waterMarkDs = mapDS.assignTimestampsAndWatermarks(
                //此处来的数据是单调递增了，不存在无序，所以不用指定为forBoundedOutOfOrderness
                WatermarkStrategy.<Tuple3<Integer, String, Integer>>forMonotonousTimestamps()
                        .withTimestampAssigner((event, timestamp) -> {
                            //指定事件时间列（需返回时间戳）
                            return TimeUtil.strToDate(event.f1).getTime();
                        })
        );
        //6.根据用户id进行分组
        KeyedStream<Tuple3<Integer, String, Integer>, Integer> keyedStream = waterMarkDs.keyBy(new KeySelector<Tuple3<Integer, String, Integer>, Integer>() {
            @Override
            public Integer getKey(Tuple3<Integer, String, Integer> input) throws Exception {
                return input.f0;
            }
        });

        //7.定义模式
        Pattern<Tuple3<Integer, String, Integer>, Tuple3<Integer, String, Integer>> pattern = Pattern.
                //定义名称
                        <Tuple3<Integer, String, Integer>>begin("firstLogin")
                .where(new SimpleCondition<Tuple3<Integer, String, Integer>>() {
                    @Override
                    public boolean filter(Tuple3<Integer, String, Integer> value) throws Exception {
                        //500代表登录失败
                        return value.f2 == 500;
                    }
                })
                //定义接下来必须发生的事。（由于下面的逻辑一样，此处也可以换为times(2).withIn()）
                .next("secondLogin").where(new SimpleCondition<Tuple3<Integer, String, Integer>>() {
                    @Override
                    public boolean filter(Tuple3<Integer, String, Integer> value) throws Exception {
                        //500代表登录失败
                        return value.f2 == 500;
                    }
                })
                //定义5s内发生以上两近邻事件
                .within(Time.seconds(5));
        //8.进行匹配检查
        PatternStream<Tuple3<Integer, String, Integer>> patternDS = CEP.pattern(keyedStream, pattern);

        //9.取出匹配检查结果,指定输出的流的格式，此处输出二维元组<Integer,String>，前者为用户id，后者为这两次登录的时间
        // select, flatSelect类似于map与flatMap
        SingleOutputStreamOperator<Tuple2<Integer, String>> selectDs = patternDS.select(new PatternSelectFunction<Tuple3<Integer, String, Integer>, Tuple2<Integer, String>>() {
            @Override
            public Tuple2<Integer, String> select(Map<String, List<Tuple3<Integer, String, Integer>>> map) throws Exception {
                //该map存储上面定义的模式名称
                String firstLoginTime = map.get("firstLogin").get(0).f1;
                String secondLoginTime = map.get("secondLogin").get(0).f1;

                return Tuple2.of(map.get("firstLogin").get(0).f0, firstLoginTime + ";" + secondLoginTime);

            }
        });

        //10.输出匹配结果
        selectDs.print("风险登录");

        try {
            environment.execute("CEP demo");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
